const amqplib = require('amqplib');

const consume = async (connect_url, in_exchange, in_queue, handler) => {
    return amqplib.connect(connect_url)
    .then(conn=>{
        return conn.createChannel()
        .then(ch=>{
            return ch.assertExchange(in_exchange, 'fanout', {'durable': true})
                    .then(_=>{
                        return ch.assertQueue(in_queue, {'exclusive': false})
                                .then(q=>{
                                    return ch.bindQueue(q.queue, in_exchange, '')
                                    .then(_=>{
                                        return ch.prefetch(2);
                                    })
                                    .then(_=>{
                                        return ch.consume(q.queue, async function(msg) {
                                            await handler(msg.content.toString());
                                            ch.ack(msg);
                                        }, {
                                            noAck: false
                                        });
                                    });   
                                });
                    }).catch(conn.close)
        });
    }).catch(err=>{
        return Promise.reject(err);
    });
}

const publish = async (connect_url, out_exchange, data) => {
    return amqplib.connect(connect_url)
    .then(conn=>{
        return conn.createConfirmChannel().then(ch=>{
            return ch.assertExchange(out_exchange, 'fanout', {'durable': true}).then(ok=>{
                if(typeof data == 'object')data = JSON.stringify(data);
                return ch.publish(out_exchange, '', new Buffer(data), {'deliveryMode': true});
            }).finally(_=>{
                ch.close();
                return Promise.resolve();
            });
        }).finally(_=>{
            conn.close();
            return Promise.resolve();
        });
    }).catch(err=>{
        return Promise.reject(err);
    });
}

exports.consume = consume;
exports.publish = publish;